#include "common.h"
/*服务端ip地址*/
char *server_ip;
/*记录poll_cq完成的次数*/
static int poll_finsh;
/*rdma相关的全局结构体定义*/
static struct rdma_info rdma_res;
/*本端的qp相关信息*/
static struct qp_info local_dest;
/*对端的qp相关信息*/
static struct qp_info remote_dest;
/*对端的mr信息*/
static struct mr_info remote_key;
/*文件信息结构体定义*/
static struct file_info file_data;
/***************************************************************
 * @file       common.c
 * @author     viscore-txj
 * @brief      client socket init
 * @param      fd
 * @version    v2
 * @return     SUCCESS/FAILURE
 * @date       2023/2/20
 **************************************************************/
int client_socket_init(int *fd)
{
    int sockfd = -1;
    struct sockaddr_in serv_addr;
    memset(&serv_addr, 0, sizeof(struct sockaddr_in));
    /*创建套接字*/
    if((sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        level_print(ERR, "Could not create socket");
        return FAILURE;
    }
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port = htons(PORT);
    serv_addr.sin_addr.s_addr = inet_addr(server_ip);
    /*对服务端发起连接*/
    if(connect(sockfd, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0) {
        level_print(ERR, "connect error");
        return FAILURE;
    }
    *fd = sockfd;
    return SUCCESS;
}
/***************************************************************
 * @file       common.c
 * @author     viscore-txj
 * @brief      获取服务端的qp和mr相关信息
 * @param      fd(输入参数)
 * @version    v2
 * @return     SUCCESS/FAILURE
 * @date       2023/2/20
 **************************************************************/
int get_server_qp_mr_info(int fd)
{
    int offset = 0;
    char *qpbuf = NULL, *mrbuf = NULL, *file_info_buf = NULL;
    /*发送客户端文件相关的信息到服务端*/
    if((file_info_buf = (char *)malloc(sizeof(struct file_info))) == NULL) {
        level_print(ERR, "file_info_buf malloc failed");
        return FAILURE;
    }
    memset(file_info_buf, 0, sizeof(struct file_info));
    memcpy(file_info_buf, &file_data, sizeof(struct file_info));
    offset = 0;
    while(offset < sizeof(struct file_info)) {
        offset += write(fd, file_info_buf + offset, sizeof(struct file_info) - offset);
        if(offset == 0 || offset < 0){
            level_print(ERR, "write file_info failed");
            goto file_buf_err;
        }
    }
    /*发送qp相关信息到对端*/
    if((qpbuf = (char *)malloc(sizeof(struct qp_info))) == NULL) {
        level_print(ERR, "qpbuf malloc failed");
        goto file_buf_err;
    }
    memset(qpbuf, 0, sizeof(struct qp_info));
    memcpy(qpbuf, &local_dest, sizeof(struct qp_info));
    offset = 0;
    while(offset < sizeof(struct qp_info)) {
        offset += write(fd, qpbuf + offset, sizeof(struct qp_info) - offset);
        if(offset == 0 || offset < 0) {
            level_print(ERR, "write qp_info failed");
            goto qp_buf_err;
        }
    }
    /*清除qpbuf里面的数据，然后开始读取对端发送的qp相关信息*/
    memset(qpbuf, 0, sizeof(struct qp_info));
    offset = 0;
    while(offset < sizeof(struct qp_info)) {
        offset += read(fd, qpbuf + offset, sizeof(struct qp_info) - offset);
        if(offset == 0 || offset < 0) {
            level_print(ERR, "read qp_info failed");
            goto qp_buf_err;
        }
    }
    /*将读取到的服务端的qp相关信息拷贝到remote_dest结构体里*/
    memcpy(&remote_dest, qpbuf, sizeof(struct qp_info));
    /*开始读取服务端的mr信息*/
    if((mrbuf = (char*)malloc(sizeof(struct mr_info))) == NULL) {
        level_print(ERR, "mrbuf malloc failed");
        goto qp_buf_err;
    }
    memset(mrbuf, 0, sizeof(struct mr_info));
    offset = 0;
    while(offset < sizeof(struct mr_info)) {
        offset += read(fd, mrbuf + offset, sizeof(struct mr_info) - offset);
        if(offset == 0 || offset < 0) {
            level_print(ERR, "read mr_info failed");
            goto mr_buf_err;
        }
    }
    memcpy(&remote_key, mrbuf, sizeof(struct mr_info));
#ifdef VISDEBUG
    /*打印读取到的对端的qp相关信息*/
    char gid[33];
    inet_ntop(AF_INET6, &remote_dest.gid, gid, sizeof(gid));
    level_print(DEBUG, "remote dest, qpn: %d, psn: %d, lid: %d, gid: %s", \
        remote_dest.qpn, remote_dest.psn, remote_dest.lid, gid);
    level_print(DEBUG, "remote addr : %ld, remote key : %d", remote_key.remote_addr, remote_key.rkey);
#endif
    return SUCCESS;
mr_buf_err:
    free(mrbuf);
qp_buf_err:
    free(qpbuf);
file_buf_err:
    free(file_info_buf);
    return FAILURE;
}
/***************************************************************
 * @file       common.c
 * @author     viscore-txj
 * @brief      获取客户端的qp相关信息
 * @param      fd(套接字fd)
 * @version    v2
 * @return     SUCCESS/FAILURE
 * @date       2023/2/20
 **************************************************************/
int get_client_qp_info(int fd)
{
    int offset = 0;
    char *qpbuf = NULL, *mrbuf = NULL, *file_info_buf = NULL;
    struct mr_info remote_key;
    memset(&remote_key, 0, sizeof(struct mr_info));
    /*读取客户端发送过来的文件信息*/
    if((file_info_buf = (char *)malloc(sizeof(struct file_info))) == NULL) {
        level_print(ERR, "qpbuf malloc failed\n");
        return FAILURE;
    }
    memset(file_info_buf, 0, sizeof(struct file_info));
    offset = 0;
    while(offset < sizeof(struct file_info)) {
        offset += read(fd, file_info_buf + offset, sizeof(struct file_info) - offset);
        if(offset == 0 || offset < 0) {
            level_print(ERR, "read file_info failed");
            goto file_buf_err;
        }
    }
    memcpy(&file_data, file_info_buf, sizeof(struct file_info));
    /*读取对端发送过来的qp相关信息*/
    if((qpbuf = (char *)malloc(sizeof(struct qp_info))) == NULL) {
        level_print(ERR, "qpbuf malloc failed");
        goto file_buf_err;
    }
    memset(qpbuf, 0, sizeof(struct qp_info));
    offset = 0;
    while (offset < sizeof(struct qp_info)) {
        offset += read(fd, qpbuf + offset, sizeof(struct qp_info) - offset);
        if(offset == 0 || offset < 0) {
            level_print(ERR, "read qp_info failed");
            goto qp_buf_err;
        }
    }
    memcpy(&remote_dest, qpbuf, sizeof(struct qp_info));
    memset(qpbuf, 0, sizeof(struct qp_info));
    /*将本地的qp相关信息拷贝到qpbuf里，然后发送到服务端*/
    memcpy(qpbuf, &local_dest, sizeof(struct qp_info));
    offset = 0;
    while(offset < sizeof(struct qp_info)) {
        offset += write(fd, qpbuf + offset, sizeof(struct qp_info) - offset);
        if(offset == 0 || offset < 0) {
            level_print(ERR, "write qp_info failed");
            goto qp_buf_err;
        }
    }
    /*将服务端的mr信息发送到客户端*/
    if((mrbuf = (char*)malloc(sizeof(struct mr_info))) == NULL) {
        level_print(ERR, "mrbuf malloc failed");
        goto qp_buf_err;
    }
    memset(mrbuf, 0, sizeof(struct mr_info));
    remote_key.remote_addr = (uint64_t)rdma_res.mrbuf;
    remote_key.rkey = rdma_res.mr->rkey;
    memcpy(mrbuf, &remote_key, sizeof(struct mr_info));
    offset = 0;
    while(offset < sizeof(struct mr_info)) {
        offset += write(fd, mrbuf + offset, sizeof(struct mr_info) - offset);
        if(offset == 0 || offset < 0){
            level_print(ERR, "write mr_info failed");
            goto mr_buf_err;
        }
    }
#ifdef VISDEBUG
    /*打印客户端发送过来的文件相关信息(文件大小.文件名称等)*/
    level_print(DEBUG, "file_name = %s, file_size = %ld, file_send_num = %d, copy_num = %d", \
        file_data.file_name, file_data.file_size, file_data.file_send_num, file_data.copy_num);
    /*打印对端发送过来的qp相关信息*/
    char gid[33];
    inet_ntop(AF_INET6, &remote_dest.gid, gid, sizeof(gid));
    level_print(DEBUG, "remote dest, qpn: %d, psn: %d, lid: %d, gid: %s", \
        remote_dest.qpn, remote_dest.psn, remote_dest.lid, gid);
    level_print(DEBUG, "remote addr : %ld, remote key : %d", remote_key.remote_addr, remote_key.rkey);
#endif
    return SUCCESS;
mr_buf_err:
    free(mrbuf);
qp_buf_err:
    free(qpbuf);
file_buf_err:
    free(file_info_buf);
    return FAILURE;
}
/***************************************************************
 * @file       common.c
 * @author     viscore-txj
 * @brief      server socket init
 * @param      fd(套接字fd)
 * @version    v2
 * @return     SUCCESS/FAILURE
 * @date       2023/2/18
 **************************************************************/
int server_scoket_init(int *fd)
{
    int sockfd = -1, ret = -1, opt = 1;
    struct sockaddr_in serv_addr;
    memset(&serv_addr, 0, sizeof(struct sockaddr_in));
    /*创建套接字*/
    sockfd = socket(AF_INET, SOCK_STREAM, 0);
    if(sockfd < 0) {
        level_print(ERR, "server create socket failed");
        return FAILURE;
    }
    /*设置套接字属性*/
    if((ret = setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &opt,sizeof(opt))) != 0 ) {
        level_print(ERR, "setsockopt SO_REUSEADDR failed");
        return ret;
    }
    if((ret = setsockopt(sockfd, SOL_SOCKET, SO_REUSEPORT, &opt,sizeof(opt))) != 0 ) {
        level_print(ERR, "setsockopt SO_REUSEPORT failed");
        return ret;
    }
    memset(&serv_addr, 0, sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    serv_addr.sin_port = htons(PORT);
    /*绑定*/
    if(bind(sockfd, (const struct sockaddr*)&serv_addr, sizeof(serv_addr))) {
        level_print(ERR, "bind error");
        return FAILURE;
    }
    /*这里参数给1,可以根据自己场景修改*/
    if(listen(sockfd, 1)) {
        level_print(ERR, "listen error");
        return FAILURE;
    }
    /*等待客户端发起连接*/
    *fd = accept(sockfd, (struct sockaddr*)NULL, NULL);
    if(*fd < 0) {
        level_print(ERR, "accept error");
        return FAILURE;
    }
    /*关闭监听套接字,已经不需要了*/
    close(sockfd);
    return SUCCESS;
}
/***************************************************************
 * @file       common.c
 * @author     viscore-txj
 * @brief      使用共享内存的方式申请内存
 * @param      buf_len(mr注册时的大小)
 * @version    v2
 * @return     mr_buf/NULL
 * @date       2023/2/17
 **************************************************************/
static void* alloc_hugepage_region(size_t buf_len)
{
    int huge_shmid = 0;
    void *mr_buf = NULL;
    huge_shmid = shmget(IPC_PRIVATE, buf_len, SHM_HUGETLB
        | IPC_CREAT | SHM_R | SHM_W);
    if(huge_shmid < 0) {
        level_print(ERR, "shmget failed");
        return NULL;
    }
    mr_buf = (void *)shmat(huge_shmid, SHMAT_ADDR, SHMAT_FLAGS);
    if(mr_buf == (void *)-1) {
        level_print(ERR, "shmat failed");
        return NULL;
    }
    if(shmctl(huge_shmid, IPC_RMID, 0) != 0) {
        level_print(ERR, "shmctl failed");
        return NULL;
    }
    return mr_buf;
}
/***************************************************************
 * @file       common.c
 * @author     viscore-txj
 * @brief      rdma init
 * @param      无
 * @version    v2
 * @return     SUCCESS/FAILURE
 * @date       2023/2/20
 **************************************************************/
int rdma_init()
{
    struct ibv_device **dev_list, *ib_dev;
    struct ibv_qp_init_attr qp_init_attr;
    union ibv_gid temp_gid;
    memset(&rdma_res, 0, sizeof(struct rdma_info));
    memset(&qp_init_attr, 0, sizeof(struct ibv_qp_init_attr));
    memset(&temp_gid, 0, sizeof(union ibv_gid));
    if(!server_ip)
        rdma_res.mrbuf_size = S_MR_BUFFER_SIZE;
     else
        rdma_res.mrbuf_size = C_MR_BUFFER_SIZE;
    rdma_res.mrbuf = alloc_hugepage_region(rdma_res.mrbuf_size);
    if(!rdma_res.mrbuf) {
        level_print(ERR, "alloc_hugepage_region failed");
        return FAILURE;
    }
    /*对申请的内存进行清零处理*/
    memset(rdma_res.mrbuf, 0x00, rdma_res.mrbuf_size);
    /*获取系统所有的rdma设备*/
    dev_list = ibv_get_device_list(NULL);
    if(!dev_list) {
        level_print(ERR, "Failed to get IB devices list");
        goto rdma_mr_buf_err;
    } else {
        ib_dev = dev_list[0];
        if (!ib_dev) {
            level_print(ERR, "No IB devices found");
            goto rdma_mr_buf_err;
        }
    }
    /*释放设备列表,这个结构体到这里已经不需要了*/
    ibv_free_device_list(dev_list);
    /*打开设备*/
    rdma_res.ctx = ibv_open_device(ib_dev);
    if(!rdma_res.ctx) {
        level_print(ERR, "Couldn't get context for %s", ibv_get_device_name(ib_dev));
        goto rdma_mr_buf_err;
    }
    /*保护域pd注册*/
    rdma_res.pd = ibv_alloc_pd(rdma_res.ctx);
    if(!rdma_res.pd) {
        level_print(ERR, "Failed to allocate Protection Domain");
        goto rdma_ctx_err;
    }
    /*mr注册*/
    rdma_res.mr = ibv_reg_mr(rdma_res.pd,
            rdma_res.mrbuf, rdma_res.mrbuf_size,
            IBV_ACCESS_LOCAL_WRITE |
            IBV_ACCESS_REMOTE_READ |
            IBV_ACCESS_REMOTE_WRITE);
    if(!rdma_res.mr) {
        level_print(ERR, "Couldn't register Memory Region");
        goto rdma_pd_err;
    }
    /*创建cq,队列深度为COUNT * 2(send+recv)*/
    rdma_res.cq = ibv_create_cq(rdma_res.ctx, COUNT * 2, NULL, NULL, 0);
    if(!rdma_res.cq) {
        level_print(ERR, "Couldn't create Completion Queue");
        goto rdma_mr_err;
    }
    /*设置qp属性*/
    qp_init_attr.send_cq = rdma_res.cq;
    qp_init_attr.recv_cq = rdma_res.cq;
    qp_init_attr.cap.max_send_wr = COUNT;
    qp_init_attr.cap.max_recv_wr = COUNT;
    qp_init_attr.cap.max_send_sge = 1;
    qp_init_attr.cap.max_recv_sge = 1;
    qp_init_attr.qp_type = IBV_QPT_RC;
    qp_init_attr.sq_sig_all = 0;
    /*创建qp*/
    rdma_res.qp = ibv_create_qp(rdma_res.pd, &qp_init_attr);
    if(!rdma_res.qp) {
        level_print(ERR, "Couldn't create Queue Pair");
        goto rdma_cq_err;
    }
    /*获取lid*/
    if(ibv_query_port(rdma_res.ctx, IB_PORT, &rdma_res.port_info)) {
        level_print(ERR, "Failed to query IB port information");
        goto rdma_cq_err;
    } else {
        local_dest.lid = rdma_res.port_info.lid;
        if(rdma_res.port_info.link_layer != IBV_LINK_LAYER_ETHERNET && !local_dest.lid) {
            level_print(ERR, "Couldn't get LID");
            goto rdma_qp_err;
        }
    }
    /*获取gid*/
    if(ibv_query_gid(rdma_res.ctx, IB_PORT, GIDX, &temp_gid)) {
        level_print(ERR, "ibv query gid failed");
        goto rdma_qp_err;
    }
    memcpy(local_dest.gid.raw, temp_gid.raw ,16);
    /*获取qp*/
    local_dest.qpn = rdma_res.qp->qp_num;
    /*获取psn*/
    local_dest.psn = lrand48() & 0xffffff;
#ifdef VISDEBUG
    char gid[33];
    level_print(DEBUG, "local rkey: 0x%06x, local lkey: 0x%06x",  rdma_res.mr->rkey, rdma_res.mr->lkey);
    level_print(DEBUG, "gid : %02x:%02X:%d:%d:%d:%d", temp_gid.raw[10],
        temp_gid.raw[11], temp_gid.raw[12], temp_gid.raw[13], temp_gid.raw[14], temp_gid.raw[15]);
    inet_ntop(AF_INET6, &local_dest.gid, gid, sizeof(gid));
    level_print(DEBUG, "local dest, qpn: %d, psn: %d, lid: %d, gid: %s", \
        local_dest.qpn, local_dest.psn, local_dest.lid, gid);
#endif
    return SUCCESS;
rdma_mr_buf_err:
    shmdt(rdma_res.mrbuf);
rdma_ctx_err:
    if(rdma_res.ctx)
        ibv_close_device(rdma_res.ctx);
rdma_pd_err:
    if(rdma_res.pd)
        ibv_dealloc_pd(rdma_res.pd);
rdma_mr_err:
    if(rdma_res.mr)
        ibv_dereg_mr(rdma_res.mr);
rdma_cq_err:
    if(rdma_res.cq)
        ibv_destroy_cq(rdma_res.cq);
rdma_qp_err:
    if(rdma_res.qp)
        ibv_destroy_qp(rdma_res.qp);
    return FAILURE;
}
/***************************************************************
 * @file       common.c
 * @author     viscore-txj
 * @brief      qp状态修改(RTR->RTS)
 * @param      无
 * @version    v2
 * @return     0/1
 * @date       2023/2/20
 **************************************************************/
int qp_state_modify()
{
    int qp_attr_mask = 0;
    struct ibv_qp_attr qp_attr;
    memset(&qp_attr, 0, sizeof(qp_attr));
    /*将qp状态修改为INIT*/
    qp_attr.qp_state = IBV_QPS_INIT;
    qp_attr.pkey_index = 0;
    qp_attr.port_num = IB_PORT;
    qp_attr.qp_access_flags = IBV_ACCESS_REMOTE_READ |
        IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_LOCAL_WRITE;
    qp_attr_mask = IBV_QP_STATE | IBV_QP_PKEY_INDEX |
        IBV_QP_PORT | IBV_QP_ACCESS_FLAGS;
    if(ibv_modify_qp(rdma_res.qp, &qp_attr, qp_attr_mask)) {
        level_print(ERR, "Failed to modify QP to INIT");
        return FAILURE;
    }
    /*将qp状态修改为RTR*/
    qp_attr_mask = 0;
    memset(&qp_attr, 0, sizeof(qp_attr));
    qp_attr.qp_state = IBV_QPS_RTR;
    qp_attr.ah_attr.port_num  = IB_PORT;
    qp_attr.ah_attr.dlid      = remote_dest.lid;
    qp_attr.ah_attr.sl        = 0;
    qp_attr.ah_attr.src_path_bits = 0;
    qp_attr.path_mtu       = IBV_MTU_1024,
    qp_attr.dest_qp_num    = remote_dest.qpn,
    qp_attr.rq_psn         = remote_dest.psn,
    qp_attr.max_dest_rd_atomic = 1;
    qp_attr.min_rnr_timer = 12;
    if(GIDX >= 0) {
        qp_attr.ah_attr.is_global  = 1;/*该值是1表明grh的设置是有效的*/
        /*以下字段用于设置Roce v2的协议头*/
        qp_attr.ah_attr.grh.flow_label = 0;
        qp_attr.ah_attr.grh.hop_limit = 0xFF;
        qp_attr.ah_attr.grh.sgid_index = GIDX;
        qp_attr.ah_attr.grh.traffic_class = 0;
        qp_attr.ah_attr.grh.dgid = remote_dest.gid;
    } else {
        /*该值设置为0说明grh的设置是无效的*/
        qp_attr.ah_attr.is_global = 0;
    }
    qp_attr_mask = IBV_QP_STATE | IBV_QP_AV |
        IBV_QP_MAX_DEST_RD_ATOMIC | IBV_QP_MIN_RNR_TIMER |
        IBV_QP_PATH_MTU | IBV_QP_DEST_QPN | IBV_QP_RQ_PSN;
    if(ibv_modify_qp(rdma_res.qp, &qp_attr, qp_attr_mask) != 0) {
        level_print(ERR, "Failed to modify QP to RTR");
        return FAILURE;
    }
    /*将qp状态修改为RTS*/
    qp_attr_mask = 0;
    memset(&qp_attr, 0, sizeof(qp_attr));
    qp_attr_mask = IBV_QP_STATE | IBV_QP_SQ_PSN |
            IBV_QP_TIMEOUT   | IBV_QP_RETRY_CNT |
            IBV_QP_RNR_RETRY | IBV_QP_MAX_QP_RD_ATOMIC;
    qp_attr.qp_state    = IBV_QPS_RTS;
    qp_attr.timeout     = 14;/*等待对端回复ack的时间*/
    qp_attr.retry_cnt   = 7;/*对端没有回复ack的话，包重传的次数，3bit*/
    qp_attr.rnr_retry   = 7;/*nack,3bit*/
    qp_attr.sq_psn      = local_dest.psn;/*包的序列号*/
    qp_attr.max_rd_atomic  = 1;
    if(ibv_modify_qp(rdma_res.qp, &qp_attr, qp_attr_mask) != 0) {
        level_print(ERR, "Failed to modify QP to RTS");
        return FAILURE;
    }
    return SUCCESS;
}
/***************************************************************
 * @file       common.c
 * @author     viscore-txj
 * @brief      获取文件名称和大小
 * @param      argv
 * @version    v2
 * @return     0(success)/other(fail)
 * @date       2023/2/20
 **************************************************************/
int get_file_info(char **argv)
{
    int ret = 1;
    char *file_name = NULL;
    struct stat s_buf;
    memset(&s_buf, 0, sizeof(struct stat));
    /*获取文件名*/
    file_name = basename(strdup(argv[2]));
    if(file_name == NULL) {
        level_print(ERR, "get file name error");
        return FAILURE;
    }
    memcpy(file_data.file_name, file_name, strlen(file_name));
    free(file_name);
    if((ret = stat(file_data.file_name, &s_buf) < 0)) {
        level_print(ERR, "stat error");
        return ret;
    }
    file_data.fp = fopen(argv[2], "r");
    if(file_data.fp == NULL) {
        level_print(ERR, "fopen file error");
        return FAILURE;
    }
    file_data.file_size = s_buf.st_size;
    if((file_data.file_size == 0) || (file_data.file_size > ((uint64_t)S_MR_BUFFER_SIZE * 5))) {
        level_print(ERR, "  file_size error<-->support file_szie [1B-5G]  ");
        fclose(file_data.fp);
        return FAILURE;
    }
    file_data.file_send_num = ((file_data.file_size % (BURST * BUFFER_SIZE_512KB) == 0) ? \
            file_data.file_size / (BURST * BUFFER_SIZE_512KB) : \
            ((int)file_data.file_size / (BURST * BUFFER_SIZE_512KB)) + 1);
    file_data.copy_num = ((file_data.file_size % (S_MR_BUFFER_SIZE) == 0) ? \
            (file_data.file_size / (S_MR_BUFFER_SIZE)) : \
            ((int)file_data.file_size / (S_MR_BUFFER_SIZE)) + 1);
#ifdef VISDEBUG
    level_print(DEBUG, "file_name = %s, file_size = %ld, file_send_num = %d, copy_num = %d", \
        file_data.file_name, file_data.file_size, file_data.file_send_num, file_data.copy_num);
#endif
    return SUCCESS;
}
/***************************************************************
 * @file       common.c
 * @author     viscore-txj
 * @brief      poll_cq(判断数据是否发送成功)
 * @param      无(NULL)
 * @version    v2
 * @return     无(NULL)
 * @date       2023/2/20
 **************************************************************/
void* poll_cq(void *arg)
{
    int n = 0, i = 0, count = 0;
    struct ibv_wc wc[BURST];
    unsigned long start_time_msec;
    unsigned long cur_time_msec;
    struct timeval cur_time;
    memset(&cur_time, 0, sizeof(struct timeval));
    memset(&wc, 0, sizeof(struct ibv_wc) * BURST);
    /*需要poll_cq的次数*/
    count = file_data.file_send_num;
    while(count != poll_finsh) {
        do {
            gettimeofday(&cur_time, NULL);
            start_time_msec = (cur_time.tv_sec * 1000) + (cur_time.tv_usec / 1000);
            n = ibv_poll_cq(rdma_res.cq, BURST, wc);
            gettimeofday(&cur_time, NULL);
            cur_time_msec = (cur_time.tv_sec * 1000) + (cur_time.tv_usec / 1000);
            if(n < 0) {
                level_print(ERR, "ibv_poll_cq error");
                return NULL;
            }
        } while((n == 0) && ((cur_time_msec - start_time_msec) < MAX_POLL_CQ_TIMEOUT));
        for(i = 0; i < n; i++) {
            if(wc[i].status != IBV_WC_SUCCESS) {
                level_print(ERR, "wc[%d].status = %d, wc[%d].wr_id = %ld", \
                    i, wc[i].status, i, wc[i].wr_id);
                pthread_exit(NULL);
            }
            poll_finsh++;
#ifdef VISDEBUG
    level_print(DEBUG, "poll_cq done, finsh = %d, wc[%d].wr_id = %ld", poll_finsh, i, wc[i].wr_id);
#endif
        }
    }
    return NULL;
}
/***************************************************************
 * @file       common.c
 * @author     viscore-txj
 * @brief      post_send
 * @param      remote_mr(buf起始地址和key),文件大小
 * @version    v2
 * @return     ibv_post_send执行结果
 * @date       2023/2/20
 **************************************************************/
int post_send_write(struct mr_info remote_mr, int sge_length, int file_send_num)
{
    int i = 0;
    static int wr_id = 0;
    static int local_index = 0, remote_index = 0;
    struct ibv_sge sge_list[BURST];
    struct ibv_send_wr wr[BURST];
    struct ibv_send_wr *bad_wr;
    memset(&sge_list, 0, sizeof(struct ibv_sge) * BURST);
    memset(&wr, 0, sizeof(struct ibv_send_wr) * BURST);
    for(i = 0; i < BURST; i++) {
        if(local_index == (C_MR_BUFFER_SIZE / BUFFER_SIZE_512KB))
            local_index = 0;
        if(remote_index == (S_MR_BUFFER_SIZE / BUFFER_SIZE_512KB))
            remote_index = 0;
        wr[i].wr_id = wr_id++;
        wr[i].sg_list = &sge_list[i];
        wr[i].num_sge = 1;
        wr[i].opcode = IBV_WR_RDMA_WRITE;
        if(i != (BURST - 1)) {
            wr[i].next = &wr[i+1];
            wr[i].send_flags = 0;
        } else {
            wr[i].next = NULL;
            wr[i].send_flags = IBV_SEND_SIGNALED;
        }
        wr[i].wr.rdma.remote_addr = remote_mr.remote_addr + ((remote_index++) * BUFFER_SIZE_512KB);
        wr[i].wr.rdma.rkey = remote_mr.rkey;
        sge_list[i].lkey = rdma_res.mr->lkey;
        sge_list[i].length = BUFFER_SIZE_512KB;
        sge_list[i].addr = (uint64_t)rdma_res.mrbuf + ((local_index++) * BUFFER_SIZE_512KB);
        /*最后一次下发WQE判断读到的文件大小是多少字节*/
        if(file_send_num == (file_data.file_send_num - 1)) {
            if((sge_length > BUFFER_SIZE_512KB) && (i == (BURST -1))) {
                sge_list[i].length = (sge_length - BUFFER_SIZE_512KB);
            } else if((sge_length < BUFFER_SIZE_512KB) && (i == 0)) {
                sge_list[i].length = sge_length;
                wr[i].next = NULL;
                wr[i].send_flags = IBV_SEND_SIGNALED;
                break;
            } else if((sge_length == BUFFER_SIZE_512KB) && (i == 0)) {
                wr[i].next = NULL;
                wr[i].send_flags = IBV_SEND_SIGNALED;
                break;
            }
        }
    }
    return ibv_post_send(rdma_res.qp, &wr[0], &bad_wr);
}
/***************************************************************
 * @file       common.c
 * @author     viscore-txj
 * @brief      将文件映射到一段虚拟内存上
 * @param      文件名称,文件大小
 * @version    v2
 * @return     文件映射的内存起始地址
 * @date       2023/2/20
 **************************************************************/
void *file_mmap(char *file_name, size_t file_size)
{
    int file_fd = -1;
    file_data.mmap_buf = NULL;
    if(!file_name) {
        level_print(ERR, "You have used NULL pointer when you get file info");
        return NULL;
    }
    /*打开文件,没有则创建一个文件*/
    file_fd = open(file_name, O_RDWR | O_CREAT | O_TRUNC, 0777);
    if(file_fd < 0) {
        level_print(ERR, "open file error");
        return NULL;
    }
    /*根据发送端的文件大小,调整服务端的文件大小(与发送方文件大小一致)*/
    if(ftruncate(file_fd, file_size) < 0) {
        level_print(ERR, "ftruncate file size error");
        goto err;
    }
    /*将文件映射到一段内存上，返回这段内存的起始地址*/
    file_data.mmap_buf = mmap(NULL, file_size,
                     PROT_READ | PROT_WRITE,
                     MAP_SHARED, file_fd, 0);
    if(MAP_FAILED == file_data.mmap_buf) {
        level_print(ERR, "mmap file error");
        goto err;
    }
    return file_data.mmap_buf;
err:
    close(file_fd);
    return NULL;
}
/***************************************************************
 * @file       common.c
 * @author     viscore-txj
 * @brief      客户端work
 * @param      fd(套接字fd)
 * @version    v2
 * @return     0/1
 * @date       2023/2/20
 **************************************************************/
int client_work(int fd)
{
    uint32_t size = 0;
    int send_num = 0, ret = 0, offset = 0;
    int local_index = 0, stop_send = 0;
    char *status_code_buf = NULL;
    if((status_code_buf = (char *)malloc(sizeof(STATUS_CODE))) == NULL) {
        level_print(ERR, "status_code buf malloc failed");
        return FAILURE;
    }
    pthread_create(&rdma_res.cq_poller_thread, NULL, poll_cq, NULL);
    for(send_num = 0; send_num < file_data.file_send_num; send_num++) {
        if(local_index == WRDTH)
            local_index = 0;
        size = fread(rdma_res.mrbuf + ((local_index++) * (BUFFER_SIZE_512KB * BURST)), \
                sizeof(char), (BUFFER_SIZE_512KB * BURST), file_data.fp);
        if(size == 0) {
            if(ferror(file_data.fp))
                goto err;
        }
        if(stop_send) {
            offset = 0;
            stop_send = 0;
            while(offset < sizeof(STATUS_CODE)) {
                offset += read(fd, status_code_buf + offset, sizeof(STATUS_CODE) - offset);
                if(offset == 0 || offset < 0) {
                    level_print(ERR, "read status code error");
                    goto err;
                }
            }
        }
        ret = post_send_write(remote_key, size, send_num);
        if(ret != 0) {
            level_print(ERR, "post_write error, ret = %d", ret);
            goto err;
        }
        if(((send_num + 1) % (S_MR_BUFFER_SIZE / (BUFFER_SIZE_512KB * BURST))) == 0) {
            offset = 0;
            stop_send = 1;
            while(offset < sizeof(STATUS_CODE)) {
                offset += write(fd, status_code_buf + offset, sizeof(STATUS_CODE) - offset);
                if(offset == 0 || offset < 0) {
                    level_print(ERR, "write status_code error");
                    goto err;
                }
            }
            if(file_data.file_send_num == (S_MR_BUFFER_SIZE / (BUFFER_SIZE_512KB * BURST)))
                goto exit;
        }
    }
    offset = 0;
    while(offset < sizeof(STATUS_CODE)) {
        offset += write(fd, status_code_buf + offset, sizeof(STATUS_CODE) - offset);
        if(offset == 0 || offset < 0) {
            level_print(ERR,  "write status_code error");
            goto err;
        }
    }
exit:
    free(status_code_buf);
    fclose(file_data.fp);
    pthread_join(rdma_res.cq_poller_thread, NULL);
    level_print(INFO, "post write finsh");
    return SUCCESS;
err:
    free(status_code_buf);
    fclose(file_data.fp);
    level_print(ERR, "post write error");
    return FAILURE;
}
/***************************************************************
 * @file       common.c
 * @author     viscore-txj
 * @brief      服务端work
 * @param      fd(套接字fd)
 * @version    v2
 * @return     0/1
 * @date       2023/2/20
 **************************************************************/
int server_work(int fd)
{
    int ret = 0, offset = 0, copy_num = 0;
    char *status_code_buf = NULL;
    void *mmap_buf = NULL;
    if((status_code_buf = (char *)malloc(sizeof(STATUS_CODE))) == NULL) {
        level_print(ERR, "status_code buf malloc failed");
        return FAILURE;
    }
    /*文件的映射*/
    mmap_buf = file_mmap(file_data.file_name, file_data.file_size);
    if(mmap_buf == NULL) {
        level_print(ERR, "mapp_buf failed");
        goto err;
    }
    for(copy_num = 0; copy_num < file_data.copy_num; copy_num++) {
        offset = 0;
        while(offset < sizeof(STATUS_CODE)) {
            offset += read(fd, status_code_buf + offset, sizeof(STATUS_CODE) - offset);
            if(offset == 0 || offset < 0) {
                level_print(ERR, "read status code error");
                goto err1;
            }
        }
        if(file_data.copy_num == 1) {
            memcpy(mmap_buf, rdma_res.mrbuf, file_data.file_size);
            break;
        } else if(copy_num != (file_data.copy_num - 1)) {
            memcpy(mmap_buf  + (copy_num * S_MR_BUFFER_SIZE), rdma_res.mrbuf, \
                S_MR_BUFFER_SIZE);
        } else if(copy_num == (file_data.copy_num - 1)) {
            memcpy(mmap_buf + (copy_num * S_MR_BUFFER_SIZE), rdma_res.mrbuf, \
                file_data.file_size - (copy_num * S_MR_BUFFER_SIZE));
            goto exit;
        } else
            level_print(ERR, "copy_num = %d", copy_num);
        offset = 0;
        while(offset < sizeof(STATUS_CODE)) {
            offset += write(fd, status_code_buf + offset, sizeof(STATUS_CODE) - offset);
            if(offset == 0 || offset < 0) {
                level_print(ERR,  "write status_code error");
                goto err1;
            }
        }
    }
exit:
    free(status_code_buf);
    munmap(file_data.mmap_buf, file_data.file_size);
    level_print(INFO, "recv finsh");
    return SUCCESS;
err1:
    munmap(file_data.mmap_buf, file_data.file_size);
    level_print(ERR, "recv error");
err:
    free(status_code_buf);
    level_print(ERR, "recv error");
    return ret;
}
/***************************************************************
 * @file       common.c
 * @author     viscore-txj
 * @brief      rdma资源的销毁
 * @param      无
 * @version    v2
 * @return     无
 * @date       2023/2/18
 **************************************************************/
void rdma_destroy()
{
    /*销毁qp*/
    if(rdma_res.qp)
        ibv_destroy_qp(rdma_res.qp);
    /*销毁cq*/
    if(rdma_res.cq)
        ibv_destroy_cq(rdma_res.cq);
    /*销毁mr*/
    if(rdma_res.mr)
        ibv_dereg_mr(rdma_res.mr);
    /*销毁pd*/
    if(rdma_res.pd)
        ibv_dealloc_pd(rdma_res.pd);
    /*关闭设备*/
    if(rdma_res.ctx)
        ibv_close_device(rdma_res.ctx);
    /*释放mrbuf*/
    shmdt(rdma_res.mrbuf);
}
